#include "includes.h"
#include "structs.h"
#include "declares.h"
/* MESSAGE BUFFER SECTION */

/* about the memory management of the 
 * actual content of MSG, in the sssp
 * case, it is pair_t.
 *
 * inmsg is responsible for item destruction
 * it use igraph_vector_ptr_set_item_destructor 
 * to set a destructor and call igraph_vector_ptr_destroy_all
 * to prevent all memory leak when the run_on_fragment finishes
 * outmsg has no destructor, so it uses igraph_vector_ptr_destroy
 *  
 * workers use IN msg to create OUT msg during evaluation,
 * then its IN msg will be cleared and
 * destroyed(both the refs and contents), 
 * the OUT will be transferred to IN
 * by coordinator, then the OUT will be cleared
 * but not destroyed as the IN now keeps ref to the MSG
 * 
 * keeping the invariant make memory leak less possible
 */

/* destroy and clear refs */
void clear_in_msg(struct msg_buf_t *msg)
{
        // destroy its all of its elements with free
        igraph_vector_ptr_t *inmsg=msg->inmsg;
        // since we register a destructor for inmsg
        // igraph will do per-element-free for us
        igraph_vector_ptr_clear(inmsg);
        
#ifndef NDEBUG
        assert(igraph_vector_ptr_empty(inmsg));
#endif

}

/* clear but not destroy, using igraph_vector_ptr_resize to achieve this */
/* perhaps by not registering a destructor for outmsg we can use igraph_vector_ptr_clear as well */
/* however, never use igraph_vector_ptr_clear OR igraph_vector_ptr_free_all if a destructor is registered */

void clear_all_out_msg(struct msg_buf_t *msg)
{
        // clear outmsg but not destroy its elements
        int nfrag=msg->nfrag;

        for(int i=0;i<nfrag;++i) {
                void *ptr=igraph_vector_ptr_e(msg->outmsg, i);
                igraph_vector_ptr_resize(ptr,0);
#ifndef NDEBUG
                assert(igraph_vector_ptr_empty(ptr));
#endif
        }
}

/* clear expired msg, exchange newly generated msg */
/* this function is used in run_on_fragment, but closely */
/* related to msg_buf, so put it here */
int sync_message(struct args_t *args, int nfrag)
{ 
        // clear all IN msg
        for(int i=0;i<nfrag;++i) {
                clear_in_msg(args[i].messages);
        }

        for(int i=0;i<nfrag;++i) {
                // loop across all msgbufs
                for(int j=0;j<nfrag;++j) {
                        // loop across all msg array in a single outmsg
                       igraph_vector_ptr_t *fid_outmsg=get_out_msg_at(args[i].messages, j);
                       int size=get_out_msg_size_at(args[i].messages,j);
#ifndef NDEBUG
                       assert(fid_outmsg);
#endif
                       for(int k=0;k<size;k++) {
                               // sync_message should know nothing about the content of msg
                               void *piece_outmsg=igraph_vector_ptr_e(fid_outmsg,k);
                               add_in_msg(args[j].messages, piece_outmsg);
                       }
                }
                clear_all_out_msg(args[i].messages);
        }
        
        return 0;
}

int add_out_msg_fid(struct msg_buf_t *msg, int fid, void *pp)
{
        igraph_vector_ptr_t *ptr=igraph_vector_ptr_e(msg->outmsg, fid);

#ifndef NDEBUG
        assert(ptr);
#endif
        igraph_vector_ptr_push_back(ptr, pp);
        return 0;
}

igraph_vector_ptr_t *get_out_msg_at(struct msg_buf_t *msg , int index)
{
        return igraph_vector_ptr_e(msg->outmsg, index);
}

int add_in_msg(struct msg_buf_t *msg, void *ptr)
{
#ifndef NDEBUG
        assert(msg->inmsg);
#endif
        return igraph_vector_ptr_push_back(msg->inmsg, ptr);
}

void *get_in_msg_at(struct msg_buf_t *msg, int index)
{

        void *rs=igraph_vector_ptr_e(msg->inmsg, index);
#ifndef NDEBUG
        assert(rs);
#endif

        return rs;
}

int get_in_msg_size(struct msg_buf_t *msg)
{
        return igraph_vector_ptr_size(msg->inmsg);
}

int get_out_msg_size_at(struct msg_buf_t *msg, int fid)
{
        
        igraph_vector_ptr_t *ptr=igraph_vector_ptr_e(msg->outmsg, fid);

#ifndef NDEBUG
        assert(ptr);
#endif

        return igraph_vector_ptr_size(ptr);
}


int msg_buf_init(struct msg_buf_t *msg_buf, int nfrag)
{
        msg_buf->nfrag=nfrag;
        msg_buf->outmsg=xmalloc(sizeof(igraph_vector_ptr_t));
        msg_buf->inmsg=xmalloc(sizeof(igraph_vector_ptr_t));
        igraph_vector_ptr_init(msg_buf->outmsg, 0);
        igraph_vector_ptr_init(msg_buf->inmsg,0);

        igraph_vector_ptr_set_item_destructor(msg_buf->inmsg,grape_plugin->msg_destroy_func);
        // for automatic clear
        for(int j=0;j<nfrag;++j) {
                 
                igraph_vector_ptr_t *ptr= xmalloc(sizeof(igraph_vector_ptr_t));
                igraph_vector_ptr_init(ptr,0);
                igraph_vector_ptr_set_item_destructor(ptr, grape_plugin->msg_destroy_func);
                igraph_vector_ptr_push_back(msg_buf->outmsg, ptr);
        }
        return 0;
} 

void msg_buf_destroy(struct msg_buf_t *msg)
{
        void *out=msg->outmsg;
        void *in=msg->inmsg;

        igraph_vector_ptr_destroy_all(in);
        free(in);
        for(int i=0;i<msg->nfrag;++i) {
                igraph_vector_ptr_t *ptr=igraph_vector_ptr_e(out,i);
                igraph_vector_ptr_destroy(ptr);
                free(ptr);
        }
        free(out);
}



void print_msg_buf(FILE *s, struct msg_buf_t *msg)
{
        int nfrag=msg->nfrag;
        fprintf(s,"msg_buf_t(nfrag=%d)\ninmsg=(inmsg_size=%ld)\n",msg->nfrag,
                        igraph_vector_ptr_size(msg->inmsg));
        int size=igraph_vector_ptr_size(msg->inmsg);
        for(int i=0;i<size;++i) {
                print_pair(s,igraph_vector_ptr_e(msg->inmsg,i));
        }
        fprintf(s,"\noutmsg=(outmsg_size=%ld)\n",
                        igraph_vector_ptr_size(msg->outmsg));
        
        for(int i=0;i<nfrag;++i) {
                fprintf(s,"\nTO fid=%d\n", i);
                void *ptr=igraph_vector_ptr_e(msg->outmsg,i);
                int size=igraph_vector_ptr_size(ptr);
                for(int j=0;j<size;++j) {
                        grape_plugin->msg_print_func(s,igraph_vector_ptr_e(ptr,j));
                        /* grape_msg_print(s,igraph_vector_ptr_e(ptr,j)); */
                }
        }
        fputs("\n", s);
}
        
void print_msg_buf_all(FILE *s, struct msg_buf_t *msg)
{
        int nfrag=msg[0].nfrag;

        fputs("BUFFERS\n---------------------------------------------------------\n", s);
        for(int i=0;i<nfrag;++i) {
                fprintf(s,"MESSAGE BUFFER %d\n", i);
                print_msg_buf(s, &msg[i]);
                fputs("---------------------------------------------------------\n", s);
        }
}

